<?php

namespace App\Kuafu\Order\Command;

use App\Kuafu\CouponExt\Model\OrderCouponExtModel;
use App\Kuafu\Order\Model\OrderCouponModel;
use App\Kuafu\Order\Model\OrderDetailModel;
use App\Kuafu\Order\Model\OrderDiscountModel;
use App\Kuafu\Order\Model\OrderModel;
use App\Library\FeiShu\SendFs;
use Com\Alibaba\Otter\Canal\Protocol\EntryType;
use Com\Alibaba\Otter\Canal\Protocol\EventType;
use Com\Alibaba\Otter\Canal\Protocol\RowChange;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;
use xingwenge\canal_php\CanalClient;
use xingwenge\canal_php\CanalConnectorFactory;
use xingwenge\canal_php\Fmt;

class OrderSync extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'order:syncOrder1 {--skip=0}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '订单同步表 1 
    --skip= : 跳过的 ACK 数量';

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
    }

    /**
     * Execute the console command
     */
    public function handle()
    {
        try {

            self::sendFs('The Canal has started.');

            $skip   = $this->option('skip');
            $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);

            $host = config('database.sync.canal.host');
            $client->connect($host);
            $client->checkValid();
            $client->subscribe(1001, 'qa', 'kf_data_statistics.*'); # 修改过滤
            while (true) {
                $message = $client->getWithoutAck(100);

                if ($skip > 0) {
                    $client->ack($message->getId());
                    $skip--;
                    continue;
                }

                if ($entries = $message->getEntries()) {
                    foreach ($entries as $entry) {
//                        self::setStartLog($entry->getHeader());
//                        self::syncOrder($entry);
//                        self::setEndLog($entry->getHeader());
                        Fmt::println($entry);
                    }
                    $client->ack($message->getId());
                }

                sleep(1);
            }
        } catch (\Exception $e) {
            self::sendFs('订单同步报错了! 非捕捉的异常 [已经阻塞后续同步了!]', self::formatException($e));
        }

    }

    private static $instance = [];

    private static function getInstance($tableName)
    {
        $tableName = self::getTableName($tableName);

        if (empty(self::$instance[$tableName])) {
            self::$instance[$tableName] = self::getTableModel($tableName);
        }

        return self::$instance[$tableName];
    }

    private static function syncOrder($entry)
    {
        switch ($entry->getEntryType()) {
            case EntryType::TRANSACTIONBEGIN:
            case EntryType::TRANSACTIONEND:
                return;
                break;
        }

        $rowChange = new RowChange();
        $rowChange->mergeFromString($entry->getStoreValue());
        $evenType = $rowChange->getEventType();
        $header   = $entry->getHeader();

        $model = self::getInstance($header->getTableName());

        foreach ($rowChange->getRowDatas() as $rowData) {

//            dump($rowData);
//
//            if (empty($model)) {
//                self::sendFs('订单同步报错了! [不阻塞后续同步]', '表名找不到对应的 model 表名为, ' . $header->getTableName() .
//                    ',对应的事件类型:' . $evenType);
//                return;
//            }
//
//            switch ($evenType) {
//                // 新增
//                case EventType::INSERT:
//                    $insertData = self::fmtColumn($rowData->getAfterColumns());
//                    if ($model->find($insertData[$model->getKeyName()])) {
//                        self::sendFs('订单同步报错了! [不阻塞后续同步]', '写入 - 主键 有重复数据! 很紧急, 快查看代码问题!' . json_encode($insertData));
//                        return;
//                    }
//                    $model->insert($insertData);
//                    break;
//                // 硬删除
//                case EventType::DELETE:
//                    break;
//                // 更新
//                case EventType::UPDATE:
//                    $beforeUpdateData = self::fmtColumn($rowData->getBeforeColumns());
//                    $updateData       = self::fmtColumn($rowData->getAfterColumns());
//                    $primaryKey       = $model->getKeyName();
//                    $model->where($primaryKey, $beforeUpdateData[$primaryKey])
//                          ->update($updateData);
//                    break;
//                default:
//                    break;
//            }
        }
    }

    private static function getTableName($tableName)
    {
        $patterns = "/kf_(.*)_\d/";
        preg_match_all($patterns, $tableName, $arr);
        if (!empty($arr[1][0])) {
            return $modelName = $arr[1][0];
        }

        return '';
    }

    private static function getTableModel($modelName)
    {
        switch ($modelName) {
            case 'order':
                return new OrderModel();
            case 'order_coupon':
                return new OrderCouponModel();
            case 'order_coupon_external':
                return new OrderCouponExtModel();
            case 'order_detail':
                return new OrderDetailModel();
            case 'order_discount':
                return new OrderDiscountModel();
            default:
                return '';
        }
    }

    private static function fmtColumn($columns)
    {
        $arr = [];
        foreach ($columns as $column) {
            $arr[$column->getName()] = $column->getValue();
        }
        return $arr;
    }

    private static function sendFs($title, $content = '')
    {
        Log::channel('orderSyncSingle')->error('同步数据库错误:', ['title' => $title, 'content' => $content]);
        SendFs::newRobot()->setTitleContent('[' . date('Y-m-d H:i:s') . '] : ' . $title, $content)->send();
    }

    private static function formatException(\Exception $e)
    {
        return json_encode([
            'line'    => $e->getLine(),
            'message' => $e->getMessage(),
            'trace'   => $e->getTrace()
        ], JSON_UNESCAPED_UNICODE);

    }

    private static function setStartLog($header)
    {
        Log::channel('orderSyncSingle')->info('同步数据库开始:', [
            sprintf("================> binlog[%s : %d],name[%s,%s], eventType: %s",
                $header->getLogfileName(),
                $header->getLogfileOffset(), $header->getSchemaName(), $header->getTableName(), $header->getEventType())
        ]);
    }

    private static function setEndLog($header)
    {
        Log::channel('orderSyncSingle')->info('同步数据库结束:', [
            sprintf("================> binlog[%s : %d],name[%s,%s], eventType: %s",
                $header->getLogfileName(),
                $header->getLogfileOffset(), $header->getSchemaName(), $header->getTableName(), $header->getEventType())
        ]);
    }
}
